package sync2ck

import (
	"fmt"
	"go-zero-demo/app/datacenter/cron/internal/svc"
	"go-zero-demo/app/datacenter/cron/internal/util/timeutil"
	"math"
	"strings"
	"time"
)

type ClientV2 struct {
	SvcCtx *svc.ServiceContext
}


const InsertTable = "INSERT INTO %s "
const MysqlHost = "SELECT %s FROM mysql('%s:%d','%s','%s', '%s', '%s') %s;"
const CreateMysqlHost = "SELECT %s FROM mysql('%s:%d','%s','%s', '%s', '%s') limit 0;"
const CreateTable = "CREATE TABLE IF NOT EXISTS %s ENGINE = %s %s ORDER BY %s AS "
const OptimizeTable = "OPTIMIZE TABLE %s final;"


func NewClientV2(svcCtx *svc.ServiceContext) *ClientV2 {
	return &ClientV2{
		svcCtx,
	}
}



func (s *ClientV2) GetCKCreateTableSql(tableNameCK, engine, partition, orderFiled, tableNameMySql string) string {
	cfg := s.SvcCtx.Config.MySQL

	return fmt.Sprintf(CreateTable, tableNameCK, engine, partition, orderFiled) + fmt.Sprintf(CreateMysqlHost, "*", cfg.Host, cfg.Port, cfg.DBName, tableNameMySql, cfg.User, cfg.Password)
}
func (s *ClientV2) GetCKInsertTableSql(tableNameCK, tableNameMySql, where, columns string) string {
	cfg := s.SvcCtx.Config.MySQL
	return fmt.Sprintf(InsertTable, tableNameCK) + fmt.Sprintf(MysqlHost, columns, cfg.Host, cfg.Port, cfg.DBName, tableNameMySql, cfg.User, cfg.Password, where)
}

func (s *ClientV2) Sync2CKBase(tableName, tableNameCK, tableSql string, removeOld bool) error {
	return s.Sync2CKWithWhere(tableName, tableNameCK, tableSql, true, removeOld, "", "")
}

func (s *ClientV2) Sync2CKWithWhere(tableName, tableNameCK, tableSql string, insert, removeOld bool, whereCus, ckWhere string) error {
	var (
		err error
	)

	fmt.Println(fmt.Sprintf("------------------%s------------------", tableName))
	if err = s.doProcessAll(tableName, tableNameCK, tableSql, insert, whereCus, ckWhere); err != nil {

	}
	fmt.Println("-----------------------------------------------")

	fmt.Println("|-【Begin】判断生成表是否存在")
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("select * FROM %s limit 1", tableNameCK)); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		return ok.Error
	}

	fmt.Println("|-【Do   】重命名旧表")
	nowDate := time.Now().Format("20060102_150405")
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("RENAME TABLE %s TO %s_%s", tableName, tableName, nowDate)); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		// return ok.Error
	}
	fmt.Println("|-【Do   】重命名新表")
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("RENAME TABLE %s TO %s", tableNameCK, tableName)); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		return ok.Error
	}
	if removeOld {
		fmt.Println("|-【Do   】删除旧表")
		if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("DROP TABLE IF EXISTS %s_%s", tableName, nowDate)); ok.Error != nil {
			fmt.Println("error ", ok.Error.Error())
			return ok.Error
		}
	}
	fmt.Println("|-【Done 】重命名表")

	return err
}

func (s *ClientV2) Sync2CKWithWhereAndRename(tableName, tableNameCK, tableSql string, insert, removeOld bool, whereCus, ckWhere string, oldName, newName string) error {
	var (
		err error
	)

	fmt.Println(fmt.Sprintf("------------------%s------------------", tableName))
	if err = s.doProcessAll(tableName, tableNameCK, tableSql, insert, whereCus, ckWhere); err != nil {

	}
	fmt.Println("-----------------------------------------------")

	fmt.Println("|-【Do   】重命名新表")
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("RENAME TABLE %s TO %s", oldName, newName)); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		return ok.Error
	}

	fmt.Println("|-【Done 】重命名表")

	return err
}

func (s *ClientV2) doProcessAll(tableName, tableNameCK string, sql string, insert bool, whereCus, ckWhere string) error {
	fmt.Println("【Begin】删除表")
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("DROP TABLE IF EXISTS %s", tableNameCK)); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		return ok.Error
	}
	fmt.Println("【Done 】删除表 ")

	fmt.Println("【Begin】创建表 ")
	if ok := s.SvcCtx.ClickHouse.Exec(sql); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		return ok.Error
	}
	fmt.Println("【Done 】创建表 ")

	if !insert {
		return nil
	}

	return s.doInsertAllData(tableName, tableNameCK, whereCus, ckWhere)
}

func (s *ClientV2) doInsertAllData(tableName, tableNameCK string, whereCus, ckWhere string) error {
	var cnt int64

	if len(whereCus) > 0 {
		if ok := s.SvcCtx.MySQL.Table(tableName).Where(strings.ReplaceAll(whereCus, "where", "")).Count(&cnt); ok.Error != nil {
			fmt.Println("error ", ok.Error.Error())
			return ok.Error
		}
	} else {
		if ok := s.SvcCtx.MySQL.Table(tableName).Count(&cnt); ok.Error != nil {
			fmt.Println("error ", ok.Error.Error())
			return ok.Error
		}
	}

	batchCK := float64(1000000)
	fmt.Println("|-【Do   】总数 ", cnt)
	batchTotal := int(math.Ceil(float64(cnt) / batchCK))
	fmt.Println("|-【Do   】分批次 ", batchTotal)

	where := "limit 1000000 offset %d"
	if len(whereCus) > 0 {
		where = whereCus + " " + where
	}
	if len(ckWhere) > 0 {
		where = ckWhere + " limit 1000000 offset %d"
	}
	fmt.Println("|-【Begin】插入数据 ")
	for i := 0; i < batchTotal; i++ {
		sql := s.GetCKInsertTableSql(tableNameCK, tableName, fmt.Sprintf(where, i*int(batchCK)), "*")

		fmt.Println("|-【Doing】批次 ", i+1)
		fmt.Println("|-【Doing】sql ", sql)
		if ok := s.SvcCtx.ClickHouse.Exec(sql); ok.Error != nil {
			fmt.Println("error ", ok.Error.Error())
			return ok.Error
		}
		fmt.Println("|-【Done 】批次 ", i+1)
	}
	fmt.Println("|-【Done 】插入数据 ")

	return nil
}



func (s *ClientV2) GetCKCreateTableSqlV2(tableNameCK, engine, partition, orderFiled, tableNameMySql string) string {
	cfg := s.SvcCtx.Config.MySQL
	return fmt.Sprintf(CreateTable, tableNameCK, engine, partition, orderFiled) + fmt.Sprintf(MysqlHost, "*", cfg.Host, cfg.Port, cfg.DBName, tableNameMySql, cfg.User, cfg.Password, "limit 0")
}

func (s *ClientV2) Sync2CK(tableList []string, partition string, orderFiled string) error {
	for _, tableName := range tableList {
		tableNameCK := timeutil.GetTableNameForRand(tableName)
		tableSql := s.GetCKCreateTableSqlV2(tableNameCK, "ReplacingMergeTree", partition, orderFiled, tableName)
		err := s.Sync2CKBaseV2(tableName, tableNameCK, tableSql, true)
		if err != nil {
			return err
		}
	}
	return nil
}


func (s *ClientV2) Sync2CKBaseV2(tableName, tableNameCK, tableSql string, removeOld bool) error {
	return s.Sync2CKWithWhereV2(tableName, tableNameCK, tableSql, true, removeOld, "", "", "")
}

func (s *ClientV2) Sync2CKWithWhereV2(tableName, tableNameCK, tableSql string, insert, removeOld bool, whereCus, ckWhere string, newTableName string) error {
	var (
		err error
	)

	fmt.Println(fmt.Sprintf("------------------%s------------------", tableName))
	if err = s.doProcessAll(tableName, tableNameCK, tableSql, insert, whereCus, ckWhere); err != nil {

	}
	fmt.Println("-----------------------------------------------")

	fmt.Println("|-【Begin】判断生成表是否存在")
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("select * FROM %s limit 0", tableNameCK)); ok.Error != nil {
		fmt.Println("error ", ok.Error.Error())
		return ok.Error
	}

	fmt.Println("|-【Do   】交换新表和旧表")
	if len(newTableName) > 0 {
		tableName = newTableName
	}
	if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("EXCHANGE TABLES %s AND %s", tableNameCK, tableName)); ok.Error != nil {
		fmt.Println("------- error -----------", ok.Error.Error())
		//return ok.Error
		//s.Logger.Error(context.Background(), "EXCHANGE", zap.Any("error", ok.Error))

		fmt.Println("|-【Do   】重命名新表")
		if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("RENAME TABLE %s TO %s", tableNameCK, tableName)); ok.Error != nil {
			fmt.Println("error ", ok.Error.Error())
			return ok.Error
		}
	}
	if removeOld {
		fmt.Println("|-【Do   】删除旧表")
		if ok := s.SvcCtx.ClickHouse.Exec(fmt.Sprintf("DROP TABLE IF EXISTS %s", tableNameCK)); ok.Error != nil {
			fmt.Println("error ", ok.Error.Error())
			return ok.Error
		}
	}
	fmt.Println("|-【Done 】成功")

	return err
}


